package com.yuanhua.sim.business.consumer;

import cn.hutool.core.bean.BeanUtil;
import com.yuanhua.core.base.entity.MsgReqBO;
import com.yuanhua.core.base.entity.MsgReqBean;
import com.yuanhua.sim.business.entity.MsgContent;
import com.yuanhua.sim.business.service.MsgContentService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.AutoConfigurationPackage;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * 消息消费者，从MQ中消费消息保存到数据库做持久化！！
 * 其中重点部分是确保消息不会重复消费；
 * 如何确保消费速度赶得上
 * @author yuanhua
 */
@Component
@RocketMQMessageListener(
        topic = "IM_MSG_STORAGE",
        consumerGroup = "MSG_STORAGE_GROUP",
        selectorExpression = "*"
)
@Slf4j
public class MessageConsumer implements RocketMQListener<MsgReqBO> {

    @Autowired
    private MsgContentService msgContentService;
    @Override
    public void onMessage(MsgReqBO msg) {
        log.info("开始消费消息：{}", msg.toString());
        //查看数据库中是否有这个ID
        String mid = msg.getMid();
        MsgContent byId = msgContentService.getById(mid);
        //如果有就丢弃消息， 重复消息
        if (byId != null) {
            return;
        }
        MsgContent msgContent = BeanUtil.copyProperties(msg, MsgContent.class);
        //没有就save到数据库，save失败抛出异常重复消费
        msgContentService.save(msgContent);
        log.info("消费消息成功：{}", msg.toString());
        //继续消费下一条
        //test
    }
}
